package com.rocket.transtion;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
 * 事务模式
 */
public class ProducerTrans {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer transProducer = new TransactionMQProducer("transProducer");
        transProducer.setNamesrvAddr("192.168.126.137:9876");
        transProducer.setExecutorService(Executors.newFixedThreadPool(5));
        transProducer.setTransactionListener(new TransactionListener() {
            ConcurrentHashMap<String, LocalTransactionState> map = new ConcurrentHashMap<>();
            /*
             在这里执行本地事务
            */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("执行本地事务!");
                System.out.println("当前线程: "+Thread.currentThread().getName());
                double random = Math.random();
                LocalTransactionState state ;
                if (random<0.333){
                    System.out.println("执行事务成功!");
                    state= LocalTransactionState.COMMIT_MESSAGE;
                }else if (random<0.666){
                    System.out.println("执行回滚事务!");
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                }else {
                    System.out.println("执行未知的事务!");
                    state = LocalTransactionState.UNKNOW;
                }
                map.put(msg.getTransactionId(), state);
                return state;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("服务器正在回查消息状态");
                System.out.println("当前线程: "+Thread.currentThread().getName());
                LocalTransactionState state = map.get(msg.getTransactionId());
                if (state==null || state == LocalTransactionState.UNKNOW){
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return state;
            }
        });

        transProducer.start();
        while (true){
            System.out.print("输入消息,用逗号分隔多条消息: ");
            String[]  msgs = new Scanner(System.in).nextLine().split(",");
            for (String msg: msgs){
                Message message = new Message("Topic4", msg.getBytes());
                System.out.println("---------发送半消息-----------");
                TransactionSendResult result = transProducer.sendMessageInTransaction(message, null);
                System.out.println("事务消息发送结果: "+ result.getLocalTransactionState().name());
            }

        }
    }
}
